package cn.itcast.function;

import cn.itcast.bean.CleanBean;
import cn.itcast.bean.IndexBean;
import cn.itcast.constant.Constant;
import cn.itcast.util.DateUtil;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.windowing.RichWindowFunction;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

public class MinIndexWindowFunction extends RichWindowFunction<CleanBean, IndexBean,String, TimeWindow> {
    private MapState<String, IndexBean> indexMs=null;
    @Override
    public void open(Configuration parameters) throws Exception {
         indexMs = getRuntimeContext().getMapState(new MapStateDescriptor<String, IndexBean>(
                "indexMs", String.class, IndexBean.class
        ));
    }

    @Override
    public void apply(String s, TimeWindow timeWindow, Iterable<CleanBean> iterable, Collector<IndexBean> collector) throws Exception {
        CleanBean cleanBean=null;
        for (CleanBean bean : iterable) {
            if(cleanBean==null){
                cleanBean=bean;
            }
            if(cleanBean.getEventTime()<bean.getEventTime()){
                cleanBean=bean;
            }
            IndexBean indexBeanLast = indexMs.get(cleanBean.getSecCode());
            Long tradeVol = 0L;
            Long tradeAmt = 0L;
            if(indexBeanLast!=null){
                //(当前分钟的总成交量-上一分钟的总成交量）
                tradeVol = cleanBean.getTradeVolumn()-indexBeanLast.getTradeVolDay();
                tradeAmt = cleanBean.getTradeAmt()-indexBeanLast.getTradeAmtDay();
            }

            Long tradeTime = DateUtil.longTimeTransfer(cleanBean.getEventTime(), Constant.format_YYYYMMDDHHMMSS);

            IndexBean indexBean = new IndexBean();
            indexBean.setEventTime(cleanBean.getEventTime());
            indexBean.setIndexCode(cleanBean.getSecCode());
            indexBean.setIndexName(cleanBean.getSecName());
            indexBean.setPreClosePrice(cleanBean.getPreClosePrice());
            indexBean.setOpenPrice(cleanBean.getOpenPrice());
            indexBean.setHighPrice(cleanBean.getMaxPrice());
            indexBean.setLowPrice(cleanBean.getMinPrice());
            indexBean.setClosePrice(cleanBean.getTradePrice());
            indexBean.setTradeVol(tradeVol);
            indexBean.setTradeAmt(tradeAmt);
            indexBean.setTradeVolDay(cleanBean.getTradeVolumn());
            indexBean.setTradeAmtDay(cleanBean.getTradeAmt());
            indexBean.setTradeTime(tradeTime);
            indexBean.setSource(cleanBean.getSource());

            indexMs.put(indexBean.getIndexCode(),indexBean);
            collector.collect(indexBean);
        }
    }
}
